package com.neusoft.bd.myflink.step07;

import com.neusoft.bd.myflink.entity.WaterSensor;
import com.neusoft.bd.myflink.myfun.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Objects;

public class TopN01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 7777)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((element, ts) -> element.getTs() * 1000L)
                );

// 全窗口函数
        sensorDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))


                .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                    HashMap<Integer, Integer> vcCountMap = new HashMap<>();

                    @Override
                    public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        for (WaterSensor waterSensor : elements) {
                            Integer vc = waterSensor.getVc();
                            Integer times = vcCountMap.get(vc);
                            if (Objects.isNull(times)) {
                                times = 1;
                            } else {
                                times++;
                            }
                            vcCountMap.put(vc, times);
                        }

                        List<Tuple2<Integer, Integer>> list = new ArrayList<>();
                        vcCountMap.forEach((vc, times) -> {
                            System.out.println("Key : " + vc + "Times : " + times);
                            Tuple2<Integer, Integer> tuple2 = Tuple2.of(vc, times);
                            list.add(tuple2);
                        });
                        list.sort((t1, t2) -> t2.f1 - t1.f1);

                        if (!Objects.isNull(list.get(0))) {
                            out.collect(list.get(0).f0 + ":" + list.get(0).f1 + "");
                        }
                        if (!Objects.isNull(list.get(1))) {
                            out.collect(list.get(1).f0 + ":" + list.get(1).f1 + "");
                        }
                    }
                }).print();
        env.execute();
    }
}

class Dog {
    public void ff(int i, String name) {

    }
}
